package scheduler

import (
	"encoding/json"

	"gitee.com/wujianqiang/golibs/pkg/factory"
	"gitee.com/wujianqiang/golibs/pkg/logging"
	"github.com/nats-io/nats.go"
)

type WokerJob struct {
	WokerEvent WokerEvent
	Reply      string
}

// 负责推送任务
type Worker struct {
	WorkerTopic string
	WorkerQueue string
	NatClient   *factory.NatClient
	Events      []Event
	JobsQueue   chan *nats.Msg
}

func (w *Worker) Run() {
	//接收任务
	w.NatClient.Conn.QueueSubscribe(w.WorkerTopic, w.WorkerQueue, func(msg *nats.Msg) {
		w.JobsQueue <- msg
	})
	for {
		select {
		case msg := <-w.JobsQueue:
			{
				var work WokerEvent
				if er := json.Unmarshal(msg.Data, &work); er != nil {
					w.NatClient.ReplyData(map[string]interface{}{
						"code":  1,
						"ermsg": er.Error(),
					}, msg.Reply)
				} else {
					//处理接收的数据,数据格式{event:"del", action:"sub", payload:map[string]interface{}}
					logging.Logger.Info("worker get job:", work)
					job := &WokerJob{
						WokerEvent: work,
						Reply:      msg.Reply,
					}
					go func(job *WokerJob) {
						w.Handle(job)
					}(job)
				}
			}
		}

	}
}

func (w *Worker) Register(event ...Event) error {
	w.Events = append(w.Events, event...)
	return nil
}

func (w *Worker) Handle(work *WokerJob) {
	for _, v := range w.Events {
		if v.EventName() == work.WokerEvent.Event {
			var res interface{}
			if work.WokerEvent.Action == "sub" {
				res = v.Action(work.WokerEvent.Payload)
			} else {
				res = v.Cancel(work.WokerEvent.Payload)
			}
			logging.Logger.Info("worker reply:", res)
			w.NatClient.ReplyData(res, work.Reply)
			return
		}
	}
}
func NewWorker(config SchedulerConfig) (*Worker, error) {
	natclient, er := config.NatConf.Connected()
	if er != nil {
		return nil, er
	}
	return &Worker{
		Events:      make([]Event, 0),
		NatClient:   natclient,
		WorkerTopic: config.WorkerTopic,
		WorkerQueue: config.WorkerQueue,
		JobsQueue:   make(chan *nats.Msg, 1024),
	}, nil
}
